package com.ucex.common.rocketmq.consumer;

import java.util.Map;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

import com.ucex.common.rocketmq.MQProperties;
import com.ucex.common.rocketmq.consumer.ali.AliMQConsumerProcess;

import lombok.extern.slf4j.Slf4j;

/**
 * 消费者启动类,无需用户启动
 */
@Component
@Slf4j
public class MessageConsumerStarter implements ApplicationListener<ContextRefreshedEvent> {

    private AbstractMessageConsumerProcess messageConsumerProcess;

    @Autowired
    public MessageConsumerStarter(MQProperties properties) {
        this.messageConsumerProcess = AliMQConsumerProcess.getInstance();
        this.messageConsumerProcess.initConsumer(properties);
    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        // 通过spring获取所有实现了MessageConsumer接口的类
        Map<String, MessageConsumer> consumers = event.getApplicationContext().getBeansOfType(MessageConsumer.class);
        if (null == consumers || consumers.isEmpty()) {
            log.info("No MessageConsumer found.");
            return;
        }

        // 订阅消息
        consumers.values().stream().forEach(consumer -> {
            log.info("Registering MessageConsumer: {}, {}, {}", consumer.getTopic(), consumer.getTag(), consumer);
            messageConsumerProcess.registerListener(consumer);
        });
         
        messageConsumerProcess.start();
        log.info("Consumers started.");
    }
}
